package com.heima.schedule.service;

import com.alibaba.fastjson.JSON;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.heima.common.cache.CacheService;
import com.heima.model.schedule.dto.Task;
import com.heima.model.schedule.pojos.Taskinfo;
import com.heima.model.schedule.pojos.TaskinfoLogs;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.util.CollectionUtils;

import java.util.Date;
import java.util.List;
import java.util.Set;

@Slf4j
@Service
public class TaskServiceImpl implements TaskService {

    @Autowired
    private TaskinfoService taskinfoService;

    @Autowired
    private TaskinfoLogsService taskinfoLogsService;

    @Autowired
    private CacheService cacheService;

    @Override
    @Transactional(rollbackFor = RuntimeException.class)
    public Long addTask(Task task) {

        //0.入参判空
        if (task == null) {
            log.warn("任务不能为空");
            return null;
        }

        //1.插入taskinfo表
        Taskinfo taskinfo = new Taskinfo();
        BeanUtils.copyProperties(task,taskinfo);
        taskinfo.setExecuteTime(new Date(task.getExecuteTime()));

        //能调Service就调Service,不用mapper
        boolean taskInfoResult = taskinfoService.save(taskinfo);
        if (!taskInfoResult) {
            log.warn("任务表数据插入失败");
            return null;
        }

        //2.插入taskinfo_log表
        TaskinfoLogs taskinfoLogs = new TaskinfoLogs();
        BeanUtils.copyProperties(task,taskinfoLogs);
        taskinfoLogs.setExecuteTime(new Date(task.getExecuteTime()));
        taskinfoLogs.setStatus(0);

        //能调Service就调Service,不用mapper
        boolean logsResult = taskinfoLogsService.save(taskinfoLogs);
        if (!logsResult) {
            log.warn("taskinfo_log表插入数据失败");
            throw new RuntimeException("askinfo_log表插入数据失败");
        }

        //3.判断任务执行时间是否小于当前时间
        //任务执行时间
        long executeTime = task.getExecuteTime();
        long currentTime = System.currentTimeMillis();
        if (executeTime <= currentTime) {
            //小于当前时间则加入当前执行队列TOPIC
            //lLeftPush("队列名称",将对象转为JSON字符串传入),数据结构为list
            cacheService.lLeftPush("TOPIC", JSON.toJSONString(task));
        }

        //4.判断任务时间是否小于五分钟
        long futureTime = currentTime + (5 * 60 * 1000);
        if (executeTime <= futureTime) {
            //小于则加入FUTURE执行队列,数据结构为zSet
            cacheService.zAdd("FUTURE", JSON.toJSONString(task),executeTime);
        }

        //5.返回任务id
        return taskinfo.getTaskId();
    }

    @Override
    @Transactional(rollbackFor = RuntimeException.class)
    public Boolean cancelTask(Long taskId) {

        if (taskId == null) {
            log.warn("传入参数为空");
            return false;
        }

        //1.删除taskinfo表中的对应信息
        boolean taskinfoResult = taskinfoService.removeById(taskId);
        if (!taskinfoResult) {
            log.warn("loginfo表删除信息失败");
            return false;
        }

        //2.更新taskinfo_log表中信息
        LambdaQueryWrapper<TaskinfoLogs> wrapper = new LambdaQueryWrapper<>();
        wrapper.eq(TaskinfoLogs::getTaskId,taskId);
        TaskinfoLogs logs = taskinfoLogsService.getOne(wrapper);
        logs.setStatus(2);
        boolean logResult = taskinfoLogsService.updateById(logs);
        //多个增删改要记得使用事务
        if (!logResult) {
            log.warn("taskinfo_log表信息更新失败");
            throw new RuntimeException("taskinfo_log表信息更新失败");
        }

        //3.创建一个与Redis中相同的Task对象，进行操作删除Redis中内容
        //Redis中是task对象，传入的也得是task对象
        Task task = new Task();
        //这里使用TaskinfoLogs的logs进行复制要注意两者的executeTime的类型不同
        BeanUtils.copyProperties(logs,task);
        task.setExecuteTime(logs.getExecuteTime().getTime());

        cacheService.lRemove("TOPIC",0,JSON.toJSONString(task));
        cacheService.zRemove("FUTURE",JSON.toJSONString(task));

        return true;
    }

    @Override
    @Transactional(rollbackFor = RuntimeException.class)
    public Boolean pollTask() {

        //1.只对topic队列进行执行,使用pop弹出方法,得到一个Task结构的JSON字符串
        //在新增任务时使用的是cacheService.lLeftPush(),取出时要注意使用对应方向的方法
        String topic = cacheService.lRightPop("TOPIC");
        if (StringUtils.isEmpty(topic)) {
            log.warn("没有从队列中得到数据");
            return false;
        }

        //2.把得到的JSON字符串转成Tas对象,JSON.parseObject(字符串,Task.class)
        Task task = JSON.parseObject(topic, Task.class);
        if (task == null) {
            log.warn("得到对象为空");
            return false;
        }

        //3.删除taskinfo表中相关信息
        boolean taskInfoResult = taskinfoService.removeById(task.getTaskId());
        if (!taskInfoResult) {
            //这里不用事务的原因是事务无法处理Redis中的操作
            log.warn("删除taskinfo表中信息失败");
            return false;
        }

        //4.更新log表中信息
        TaskinfoLogs logs = new TaskinfoLogs();
        BeanUtils.copyProperties(task,logs);
        //注意两个对象时间格式不同
        logs.setExecuteTime(new Date(task.getExecuteTime()));
        logs.setStatus(1);
        boolean logResult = taskinfoLogsService.updateById(logs);
        if (!logResult) {
            log.warn("日志表更新失败");
            throw new RuntimeException("日志表更新失败");
        }

        return true;
    }

    //使用@Scheduled注解实现定时任务
    @Override
    @Scheduled(fixedRate = 1000 * 60 * 5)
    public void refresh() {
        log.info("执行队列数据更新");

        //两个参数:"自定义锁的名称",自动释放锁的时间(毫秒数)
        String lockToken = cacheService.tryLock("refresh_lock", 5000);
        if (StringUtils.isEmpty(lockToken)) {
            //如果token为空,则证明是上锁了,终止执行
            return;
        }

        //从Redis中获取队列的数据,从一个时间段中得到一个set集合
        //3个参数:队列名称,时间起始值,时间结束值
        Set<String> future = cacheService.zRangeByScore("FUTURE", 0, System.currentTimeMillis());
        if (CollectionUtils.isEmpty(future)) {
            log.warn("数据为空无法执行");
            return;
        }

        //通过管道更新队列数据,3个参数
        //更新数据的来源队列,目标队列,更新数据
        cacheService.refreshWithPipeline("FUTURE","TOPIC",future);
    }

    @Scheduled(fixedRate = 5 * 60 * 1000)
    public void initData(){
        //1.清除缓存
        clear();

        //2.查询taskinfo表中的数据(即为还未执行的任务
        List<Taskinfo> list = taskinfoService.list();
        if (CollectionUtils.isEmpty(list)) {
            log.warn("没有待执行的任务");
            return;
        }

        for (Taskinfo taskinfo : list) {
            if (taskinfo == null) {
                //为空则跳过
                continue;
            }
            //得到任务的时间,为任务分流做准备
            long executeTime = taskinfo.getExecuteTime().getTime();
            long currentTime = System.currentTimeMillis();
            long futureTime = currentTime + (5 * 60 * 1000);

            Task task = new Task();
            BeanUtils.copyProperties(taskinfo,task);
            task.setExecuteTime(executeTime);

            if (executeTime <= currentTime) {
                //进入执行队列
                cacheService.lLeftPush("TOPIC", JSON.toJSONString(task));
            } else if (executeTime < futureTime) {
                cacheService.zAdd("FUTURE",JSON.toJSONString(task),executeTime);
            }
        }

    }
    
    public void clear(){
         cacheService.delete("TOPIC");
         cacheService.delete("FUTURE");
    }

    //todo 还有最后一部分代码没实现

}
